package org.learn.dataguru.kafka.demo.class02;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

public class ProducerTest {

	static private final String TOPIC = "test1";
	static private final String BROKER_LIST = "kafka:9092";

	public static void main(String[] args) throws InterruptedException {
	    Producer<String, String> producer = initProducer();
	    send(producer, TOPIC);
	}

	private static Producer<String, String> initProducer() {
		Properties props = new Properties();
		props.put("metadata.broker.list", BROKER_LIST);
		props.put("serializer.class", StringEncoder.class.getName());
		//props.put("partitioner.class", RandomPartitioner.class.getName());
		props.put("producer.type", "async");
		props.put("batch.num.messages", "3");
		props.put("request.required.acks", "-1");
		props.put("queue.buffer.max.ms", "10000000");
		props.put("queue.buffering.max.messages", "1000000");
		props.put("queue.enqueue.timeout.ms", "20000000");

		ProducerConfig config = new ProducerConfig(props);
		Producer<String, String> producer = new Producer<String, String>(config);
		return producer;
	}

	public static void send(Producer<String, String> producer, String topic)throws InterruptedException {
		KeyedMessage<String, String> message1 = new KeyedMessage<String, String>(topic, "1", "test 1");
		producer.send(message1);
		KeyedMessage<String, String> message2 = new KeyedMessage<String, String>(topic, "1", "test 2");
		producer.send(message2);
		KeyedMessage<String, String> message3 = new KeyedMessage<String, String>(topic, "1", "test 3");
		producer.send(message3);
		KeyedMessage<String, String> message4 = new KeyedMessage<String, String>(topic, "1", "test 4");
		producer.send(message4);
		KeyedMessage<String, String> message5 = new KeyedMessage<String, String>(topic, "1", "test 5");
		producer.send(message5);
		KeyedMessage<String, String> message6 = new KeyedMessage<String, String>(topic, "1", "test 6");
		producer.send(message6);
		KeyedMessage<String, String> message7 = new KeyedMessage<String, String>(topic, "1", "test 7");
		producer.send(message7);
		KeyedMessage<String, String> message8 = new KeyedMessage<String, String>(topic, "1", "test 8");
		producer.send(message8);
		KeyedMessage<String, String> message9 = new KeyedMessage<String, String>(topic, "1", "test 9");
		producer.send(message9);
		KeyedMessage<String, String> messag10 = new KeyedMessage<String, String>(topic, "1", "test 10");
		producer.send(messag10);
		producer.close();
	}

}
